package com.o2o.repository;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;

import javax.annotation.Resource;

import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Repository;

import com.o2o.base.BaseRepository;
import com.o2o.content.HBaseContent;
import com.o2o.model.topic.CategoryTopic;
import com.o2o.model.topic.Product;
import com.o2o.model.topic.Topic;
import com.o2o.model.topic.TopicContent;
import com.o2o.model.user.User;
import com.o2o.util.HBaseUtil;

@Scope("prototype")
@Repository
public class HBaseServiceRepository extends BaseRepository{
	
	@Resource
	HBaseRepository hBaseRepository;
	
	Map<String,String> columnMap =Collections.synchronizedMap(new HashMap<String,String>()); 
	
	/**
	 * 保存一个主题帖
	 * @param uid 用户id
	 * @param topicId 主题帖id
	 * @param userName 用户名称
	 * @param topicTitle 主题帖title
	 * @param topicString 主题帖的楼主说
	 * @param productId 商品id
	 * @throws IOException 
	 */
	public void saveAndUpdateTopic(final long topicId,final int categoryNumber,final long uid,final String userName,final String topicTitle,final String topicString,final long productId,final int browseCount,final int contentCount,final int userCount,final int greatCount,final int status,final long time) throws IOException{
		columnMap.clear();
		columnMap.put(HBaseContent.TOPIC_ID, topicId+"");
		columnMap.put(HBaseContent.TOPIC_UID, uid+"");
		columnMap.put(HBaseContent.TOPIC_USER_NAME, userName);
		columnMap.put(HBaseContent.TOPIC_STRING, topicString);
		columnMap.put(HBaseContent.TOPIC_TITLE, topicTitle);
		columnMap.put(HBaseContent.TOPIC_PRODUCT_ID, productId+"");
		columnMap.put(HBaseContent.TIMESTAMP, time+"");
		columnMap.put(HBaseContent.TOPIC_STATUS, status+"");
		columnMap.put(HBaseContent.TOPIC_BROWSE_COUNT, browseCount+"");
		columnMap.put(HBaseContent.TOPIC_USER_COUNT,userCount+"");
		columnMap.put(HBaseContent.TOPIC_GREAT_COUNT, greatCount+"");
		columnMap.put(HBaseContent.TOPIC_CONTENT_COUNT, contentCount+"");
		
		String rowKey = topicId+"";
		hBaseRepository.insertAndUpdate(HBaseContent.HBASE_TABLE_TOPIC, rowKey, HBaseContent.HBASE_FAMILY_TOPIC, columnMap);
	}
	/**
	 * 保存主题帖topic和cateogyrNumber关系，包括product,categoryNumber,topicId,time字段，
	 * @param topicId 主题帖id
	 * @param productId 商品id
	 * @param categoryNumber 分类number
	 * @param time 添加时间
	 * @throws IOException
	 */
	public void saveTopicCategoryNumber(long topicId,long productId,int categoryNumber,int productType,long time) throws IOException{
		columnMap.clear();
		columnMap.put(HBaseContent.TOPIC_CATEGORY_TOPIC_ID, topicId+"");
		columnMap.put(HBaseContent.TOPIC_CATEGORY_PRODUCT_ID, productId+"");
		columnMap.put(HBaseContent.TOPIC_CATEGORY_NUMBER, categoryNumber+"");
		columnMap.put(HBaseContent.TOPIC_PRODUCT_TYPE, productType+"");
		columnMap.put(HBaseContent.TIMESTAMP, time+"");
		//主键是categoryNumber+time desc +topicId
		long sortKey = Integer.MAX_VALUE - topicId;
		String number = HBaseUtil.numberChange(10, sortKey);
		String rowKey = ""+productType+categoryNumber+number;
		hBaseRepository.insertAndUpdate(HBaseContent.HBASE_TABLE_TOPIC_CATEGORY, rowKey, HBaseContent.HBASE_FAMILY_TOPIC_CATEGORY, columnMap);
	}
	/**
	 * 根据categoryNumber获取所有对应的主题帖对象，如果下一页，则返回的条目数会比pageSize多1
	 * @param categoryNumber 分类number
	 * @param productType 商品类型
	 * @param page 页码
	 * @param startRowKey 开始扫描的rowKey，若为null，或者当前是第一页，则是默认的开始位置
	 * @param stopRowKey 结束扫描的rowKey，同上
	 * @param pageSize 每页数量
	 * @param rowFlag boolean，表示该给定的startKey是否是精确的开始扫描的值，如果不是，则是默认的开始扫描的位置
	 * @return List<Topic>
	 * @throws IOException 
	 */
	public List<Topic> getTopicListByCategoryNumber(int categoryNumber,int productType,int page,int pageSize,String startRowKey,String stopRowKey,boolean rowFlag) throws IOException {
		if(page ==1 || startRowKey == null){//第一页或者是没有缓存的情况下
			startRowKey = ""+productType+categoryNumber+"0000000000";
		}
		if(page ==1 || stopRowKey == null){
			String tmp = HBaseUtil.getStopRowKey(categoryNumber+"");
			stopRowKey = ""+productType+tmp+"0000000000";
		}
		List<NavigableMap<byte[], byte[]>> columnKeyValueList = hBaseRepository.scanColumnKeyValueList(HBaseContent.HBASE_TABLE_TOPIC_CATEGORY, HBaseContent.HBASE_FAMILY_TOPIC_CATEGORY, startRowKey, stopRowKey,page, pageSize,rowFlag);
		int i = 0;
		if(columnKeyValueList != null && !columnKeyValueList.isEmpty()){
			String[] ids = new String[columnKeyValueList.size()];
			for (NavigableMap<byte[], byte[]> navigableMap : columnKeyValueList) {
				byte[] bs = navigableMap.get(HBaseContent.TOPIC_CATEGORY_TOPIC_ID.getBytes());
				long topicId = bs == null ? 0 :Long.parseLong(new String(bs));
				ids[i++]=topicId+"";
			}
			List<Topic> topicList = this.getTopicList(ids);
			return topicList;
		}else return null;
	}

	/**
	 * 保存一个主题帖的商品
	 * @param productName 商品名称
	 * @param productId 商品id
	 * @param productMarketPrice 商品市场价
	 * @param productCategoryNumber 商品分类number
	 * @param productPicturePath 商品主图的路径
	 * @param productBrand 商品品牌
	 * @param productType 商品的类型，1为标准品，2为本地生活服务，3为二手商品
	 * @param productDescription 商品描述
	 * @param uid 用户id
	 * @return 序列化后的id,若保存失败，则返回0
	 * @throws IOException 
	 */
	public void saveAndUpdateProduct(final long productId,final String productName,final float productPrice,final int productCategoryNumber,final String productPicture,final String productBrand,final int productType,final String productDescription,final long uid,final int saleNum,final int grade,final long time) throws IOException{
		columnMap.clear();
		columnMap.put(HBaseContent.PRODUCT_ID, productId+"");
		columnMap.put(HBaseContent.PRODUCT_NAME, productName);
		columnMap.put(HBaseContent.PRODUCT_PRICE, productPrice+"");
		columnMap.put(HBaseContent.PRODUCT_STATUS,"1");
		columnMap.put(HBaseContent.PRODUCT_CATEGORY_NUMBER, productCategoryNumber+"");
		columnMap.put(HBaseContent.PRODUCT_MAIN_PICTURE, productPicture);
		columnMap.put(HBaseContent.PRODUCT_BRAND, productBrand);
		columnMap.put(HBaseContent.PRODUCT_TYPE, productType+"");
		columnMap.put(HBaseContent.PRODUCT_DESCRIPTION, productDescription);
		columnMap.put(HBaseContent.PRODUCT_UID, uid+"");
		columnMap.put(HBaseContent.PRODUCT_SALE_NUM, saleNum+"");
		columnMap.put(HBaseContent.PRODUCT_GRADE, grade+"");
		columnMap.put(HBaseContent.TIMESTAMP, time+"");
		long rowKey = productId;
		hBaseRepository.insertAndUpdate(HBaseContent.HBASE_TABLE_PRODUCT,rowKey+"",HBaseContent.HBASE_FAMILY_PRODUCT,columnMap);
	}
	/**
	 * 保存或更新user对象
	 * @param uid 主键id
	 * @param name 用户名
	 * @param passwd 密码，经过md5处理后的
	 * @param email email邮箱
	 * @param status 用户状态
	 * @param picture 用户头像
	 * @param time 添加时间
	 * @throws IOException
	 */
	public void saveAndUpdateUser(long uid,String name,String passwd,String email,int status,String picture,long time) throws IOException{
		columnMap.clear();
		columnMap.put(HBaseContent.USER_ID, uid+"");
		columnMap.put(HBaseContent.USER_NAME, name);
		columnMap.put(HBaseContent.USER_PASSWD, passwd);
		columnMap.put(HBaseContent.USER_EMAIL, email);
		columnMap.put(HBaseContent.USER_STATUS,status+"");
		columnMap.put(HBaseContent.USER_PICTURE, picture);
		columnMap.put(HBaseContent.TIMESTAMP, time+"");
		long rowKey = uid;
		hBaseRepository.insertAndUpdate(HBaseContent.HBASE_TABLE_USER,rowKey+"",HBaseContent.HBASE_FAMILY_USER,columnMap);
	}
	/**
	 * 保存主题帖评论
	 * @param tid 评论id
	 * @param topicId 主题帖id
	 * @param content 内容
	 * @param uid 用户id
	 * @param great 赞的次数
	 * @param pid 回复的id
	 * @param time 添加时间
	 * @throws IOException
	 */
	public void saveTopicContent(long tid,long topicId,String content,long uid,int great,long pid,long time) throws IOException{
		columnMap.clear();
		columnMap.put(HBaseContent.CONTENT_ID, tid+"");
		columnMap.put(HBaseContent.CONTENT_TID, topicId+"");
		columnMap.put(HBaseContent.CONTENT_CONTENT, content);
		columnMap.put(HBaseContent.CONTENT_UID, uid+"");
		columnMap.put(HBaseContent.CONTENT_GREAT, great+"");
		columnMap.put(HBaseContent.CONTENT_PID, pid+"");
		columnMap.put(HBaseContent.TIMESTAMP, time+"");
		//topic评论表的rowKey，是主题帖id-时间 desc-用户id-评论的本身id desc 构成
		long sortKey = Integer.MAX_VALUE - tid;
		String number = HBaseUtil.numberChange(10, sortKey);
		String rowKey = ""+topicId+number;
		
		hBaseRepository.insertAndUpdate(HBaseContent.HBASE_TABLE_CONTENT, rowKey, HBaseContent.HBASE_FAMILY_CONTENT, columnMap);
	}
	
	
	/**
	 * 格式化CategoryTopic通过NavigableMap<byte[], byte[]>
	 * @param columnMap NavigableMap<byte[], byte[]> hbase查询后得到的name=value形式
	 * @return CategoryTopic
	 */
	private CategoryTopic formatCategoryTopic(NavigableMap<byte[], byte[]> columnMap){
		if(columnMap != null && !columnMap.isEmpty()){
			byte[] bs = columnMap.get(HBaseContent.TOPIC_CATEGORY_TOPIC_ID.getBytes());
			long topicId = bs == null ? 0 :Long.parseLong(new String(bs));
			bs = columnMap.get(HBaseContent.TOPIC_CATEGORY_NUMBER.getBytes());
			int categoryNumber = bs == null ? 0 :Integer.parseInt(new String(bs));
			bs = columnMap.get(HBaseContent.TOPIC_CATEGORY_PRODUCT_ID.getBytes());
			long productId = bs == null ? 0 :Long.parseLong(new String(bs));
			CategoryTopic categoryTopic = new CategoryTopic(Long.parseLong(""+categoryNumber+topicId), categoryNumber, topicId, productId);
			return categoryTopic;
		}else
			return null;
	}
	/**
	 * 格式化TopicNavigableMap，得到一个topic对象
	 * @param columnMap topic对象的name对应value map
	 * @return Topic
	 */
	private Topic formatTopicColumnMap(NavigableMap<byte[], byte[]> columnMap){
		if(columnMap != null && !columnMap.isEmpty()){
			try {
				byte[] bs = columnMap.get(HBaseContent.TOPIC_TITLE.getBytes());
				String topicTitle = bs == null ? "" :new String(bs,"utf8");
				bs = columnMap.get(HBaseContent.TOPIC_STRING.getBytes());
				String topicString = bs == null ? "" :new String(bs,"utf8");
				bs = columnMap.get(HBaseContent.TOPIC_UID.getBytes());
				long uid = bs == null ? 0 :Long.parseLong(new String(bs));
				bs = columnMap.get(HBaseContent.TOPIC_ID.getBytes());
				long id = bs == null ? 0 :Long.parseLong(new String(bs));
				bs = columnMap.get(HBaseContent.TOPIC_USER_NAME.getBytes());
				String userName = bs == null ? "" :new String(bs,"utf8");
				bs = columnMap.get(HBaseContent.TOPIC_PRODUCT_ID.getBytes());
				long productId = bs == null ? 0 :Long.parseLong(new String(bs));
				bs = columnMap.get(HBaseContent.TOPIC_STATUS.getBytes());
				int status = bs == null ? 0 :Integer.parseInt(new String(bs));
				bs = columnMap.get(HBaseContent.TOPIC_BROWSE_COUNT.getBytes());
				int browseCount =  bs == null ? 0 :Integer.parseInt(new String(bs));
				bs = columnMap.get(HBaseContent.TOPIC_CONTENT_COUNT.getBytes());
				int contentCount = bs == null ? 0 :Integer.parseInt(new String(bs));
				bs = columnMap.get(HBaseContent.TOPIC_USER_COUNT.getBytes());
				int userCount =  bs == null ? 0 :Integer.parseInt(new String(bs));
				bs = columnMap.get(HBaseContent.TOPIC_GREAT_COUNT.getBytes());
				int greatCount = bs == null ? 0 :Integer.parseInt(new String(bs));
				bs = columnMap.get(HBaseContent.TIMESTAMP.getBytes());
				long time = bs == null ? 0 :Long.parseLong(new String(bs));
				Topic topic = new Topic(id, topicTitle, topicString, time, uid, userName, status,browseCount, contentCount, userCount,greatCount, productId);
				return topic;
			} catch (Exception e) {
				e.printStackTrace();
				return null;
			} 
		}else
			return null;
	}
	
	/**
	 * 根据topicId获取对应的topic对象
	 * @param topicId 主题帖id
	 * @return Topic
	 * @throws IOException
	 */
	
	public Topic getTopicById(long topicId) throws IOException{
		long rowKey = topicId;
		NavigableMap<byte[], byte[]> columnMap = hBaseRepository.getColumnKeyValueMap(HBaseContent.HBASE_TABLE_TOPIC,rowKey+"",HBaseContent.HBASE_FAMILY_TOPIC);
		Topic topic = this.formatTopicColumnMap(columnMap);
		return topic;
	}
	/**
	 * 根据topic对象主键topicId []，批量获取所有对应的Topic对象
	 * @param topicIds 主键id数组
	 * @return List<Topic>
	 * @throws IOException
	 */
	public List<Topic> getTopicList(String[] topicIds) throws IOException{
		List<NavigableMap<byte[], byte[]>> columnKeyValueListMap = hBaseRepository.getColumnKeyValueListMap(HBaseContent.HBASE_TABLE_TOPIC, topicIds, HBaseContent.HBASE_FAMILY_TOPIC);
		List<Topic> topics = Collections.synchronizedList(new ArrayList<Topic>());
		for (NavigableMap<byte[], byte[]> navigableMap : columnKeyValueListMap) {
			Topic topic = this.formatTopicColumnMap(navigableMap);
			topics.add(topic);
		}
		return topics;
	}
	
	public Map<Long,Product> getProductListByIds(String[] productIds) throws IOException{
		List<NavigableMap<byte[], byte[]>> columnKeyValueListMap = hBaseRepository.getColumnKeyValueListMap(HBaseContent.HBASE_TABLE_PRODUCT, productIds, HBaseContent.HBASE_FAMILY_PRODUCT);
		Map<Long,Product> productMap= Collections.synchronizedMap(new HashMap<Long,Product>());
		for (NavigableMap<byte[], byte[]> navigableMap : columnKeyValueListMap) {
			Product product = this.formatProductKeyValueMap(navigableMap);
			productMap.put(product.getId(), product);
		}
		return productMap;
	}
	private Product formatProductKeyValueMap(NavigableMap<byte[], byte[]> columnMap){
		if(columnMap != null && !columnMap.isEmpty()){
			try {
				byte[] bs = columnMap.get(HBaseContent.PRODUCT_ID.getBytes());
				long pid = bs == null ? 0 :Long.parseLong(new String(bs));
				bs = columnMap.get(HBaseContent.PRODUCT_NAME.getBytes());
				String name = bs == null ? "" :new String(bs,"utf8");
				bs = columnMap.get(HBaseContent.PRODUCT_PRICE.getBytes());
				float price = bs == null ? 0 :Float.parseFloat(new String(bs));
				bs = columnMap.get(HBaseContent.PRODUCT_TYPE.getBytes());
				int type = bs == null ? 0 :Integer.parseInt(new String(bs));
				bs = columnMap.get(HBaseContent.PRODUCT_BRAND.getBytes());
				String brand = bs == null ? "" :new String(bs,"utf8");
				bs = columnMap.get(HBaseContent.PRODUCT_STATUS.getBytes());
				int status = bs == null ? 0 :Integer.parseInt(new String(bs));
				bs = columnMap.get(HBaseContent.PRODUCT_MAIN_PICTURE.getBytes());
				String picture = bs == null ? "" :new String(bs,"utf8");
				bs = columnMap.get(HBaseContent.PRODUCT_DESCRIPTION.getBytes());
				String description = bs == null ? "" :new String(bs,"utf8");
				bs = columnMap.get(HBaseContent.PRODUCT_CATEGORY_NUMBER.getBytes());
				int categoryNumber = bs == null ? 0 :Integer.parseInt(new String(bs));
				bs = columnMap.get(HBaseContent.PRODUCT_UID.getBytes());
				long uid = bs == null ? 0 :Long.parseLong(new String(bs));
				bs = columnMap.get(HBaseContent.PRODUCT_SALE_NUM.getBytes());
				int saleNum = bs == null ? 0 :Integer.parseInt(new String(bs));
				bs = columnMap.get(HBaseContent.PRODUCT_GRADE.getBytes());
				int grade = bs == null ? 0 :Integer.parseInt(new String(bs));
				bs = columnMap.get(HBaseContent.TIMESTAMP.getBytes());
				long time = bs == null ? 0 :Long.parseLong(new String(bs));
				Product product = new Product(pid, name, price, description, time, "", status, type, categoryNumber, brand, picture, saleNum, grade,uid);
				return product;
			} catch (Exception e) {
				e.printStackTrace();
				return null;
			} 
		}else
			return null;
	}
	
	/**
	 * 根据productid获取对应的product对象
	 * @param productId 主键id
	 * @return Product
	 * @throws IOException
	 */
	public Product getProductById(long productId) throws IOException{
		long rowKey = productId;
		NavigableMap<byte[], byte[]> columnMap = hBaseRepository.getColumnKeyValueMap(HBaseContent.HBASE_TABLE_PRODUCT,rowKey+"",HBaseContent.HBASE_FAMILY_PRODUCT);
		return this.formatProductKeyValueMap(columnMap);
	}
	/**
	 * 返回topic和product所有封装好的product对象,主体是topic
	 * @param tid 主题帖id
	 * @return Product(父类已经查询封装)
	 * @throws IOException
	 */
	public Topic getTopicAndProductBytID(long tid) throws IOException{
		Topic topic = this.getTopicById(tid);
		if(topic != null){
			long rowKey = topic.getProductId();
			NavigableMap<byte[], byte[]> columnMap = hBaseRepository.getColumnKeyValueMap(HBaseContent.HBASE_TABLE_PRODUCT,rowKey+"",HBaseContent.HBASE_FAMILY_PRODUCT);
			Product product = this.formatProductKeyValueMap(columnMap);
			topic.setProduct(product);
		}
		return topic;
	}
	private User formatUserKeyValueMap(NavigableMap<byte[], byte[]> navigableMap){
		if(navigableMap != null && !navigableMap.isEmpty()){
			byte[] bs = navigableMap.get(HBaseContent.USER_ID.getBytes());
			long uid = bs == null ? 0 :Long.parseLong(new String(bs));
			bs = navigableMap.get(HBaseContent.USER_NAME.getBytes());
			String name = bs == null ? "" :new String(bs);
			bs = navigableMap.get(HBaseContent.USER_PASSWD.getBytes());
			String passwd = bs == null ? "" :new String(bs);
			bs = navigableMap.get(HBaseContent.USER_EMAIL.getBytes());
			String email = bs == null ? "" :new String(bs);
			bs = navigableMap.get(HBaseContent.USER_STATUS.getBytes());
			int status = bs == null ? 0 :Integer.parseInt(new String(bs));
			bs = navigableMap.get(HBaseContent.USER_PICTURE.getBytes());
			String picture = bs == null ? "" :new String(bs);
			bs = navigableMap.get(HBaseContent.TIMESTAMP.getBytes());
			long time = bs == null ? 0 :Long.parseLong(new String(bs));
			User user = new User(uid, name, passwd, email, status, picture,time);
			return user;
		}else return null;
		
	}
	/**
	 * 根据用户id数组，批量获取user对象
	 * @param ids long[] 主键id
	 * @return List<User>
	 * @throws IOException 
	 */
	public Map<Long,User> getUserListByIds(String[] userIds) throws IOException{
		List<NavigableMap<byte[], byte[]>> columnKeyValueListMap = hBaseRepository.getColumnKeyValueListMap(HBaseContent.HBASE_TABLE_USER, userIds, HBaseContent.HBASE_FAMILY_USER);
		Map<Long,User> userMap = Collections.synchronizedMap(new HashMap<Long,User>());
		for (NavigableMap<byte[], byte[]> navigableMap : columnKeyValueListMap) {
			User user = this.formatUserKeyValueMap(navigableMap);
			userMap.put(user.getId(), user);
		}
		return userMap;
	}
	
	/**
	 * 根据id获取User对象
	 * @param userId 用户id
	 * @return User
	 * @throws IOException
	 */
	public User getUserById(long userId) throws IOException{
		long rowKey = userId;
		NavigableMap<byte[], byte[]> columnMap = hBaseRepository.getColumnKeyValueMap(HBaseContent.HBASE_TABLE_USER,rowKey+"",HBaseContent.HBASE_FAMILY_USER);
		User user = this.formatUserKeyValueMap(columnMap);
		return user;
	}
	/**
	 * 根据主题帖id获取评论回复List，如果下一页，则返回的条目数会比pageSize多1
	 * @param topicId 主题帖id
	 * @param page 页码
	 * @param pageSize 每页数量
	 * @param startRowKey 服务器端缓存的开始扫描的rowKey，若为null,或者当前page=1则从默认位置开始 
	 * @param stopRowKey 缓存的结束位置的rowKey，同startRowKey
	 * @param rowFlag boolean，表示该给定的startKey是否是精确的开始扫描的值，如果不是，则是默认的开始扫描的位置
	 * @return List<TopicContent>
	 * @throws IOException
	 */
	public List<TopicContent> getTopicContentListByTid(long topicId,int page,int pageSize,String startRowKey,String stopRowKey,boolean rowFlag) throws IOException{
		if(page ==1 || startRowKey== null){//第一页或者是没有缓存的情况下
			startRowKey = ""+topicId+"0000000000";//补齐位
		}
		if(page ==1 || stopRowKey== null){
			stopRowKey = topicId+1+"0000000000";
		}
		List<NavigableMap<byte[], byte[]>> columnKeyValueListMap = hBaseRepository.scanColumnKeyValueList(HBaseContent.HBASE_TABLE_CONTENT, HBaseContent.HBASE_FAMILY_CONTENT, startRowKey, stopRowKey,page, pageSize,rowFlag);
		List<TopicContent> contents = Collections.synchronizedList(new ArrayList<TopicContent>());
		if(columnKeyValueListMap != null){
			for (NavigableMap<byte[], byte[]> navigableMap : columnKeyValueListMap) {
				TopicContent content = this.formatTopicContentKeyValueMap(navigableMap);
				contents.add(content);
			}
		}
		return contents;
	}
	private TopicContent formatTopicContentKeyValueMap(NavigableMap<byte[], byte[]> columnMap){
		if(columnMap != null && !columnMap.isEmpty()){
			try {
				byte[] bs = columnMap.get(HBaseContent.CONTENT_ID.getBytes());
				long contentId = bs == null ? 0 :Long.parseLong(new String(bs));
				bs = columnMap.get(HBaseContent.CONTENT_CONTENT.getBytes());
				String contentString = bs == null ? "" :new String(bs,"utf-8");
				bs = columnMap.get(HBaseContent.CONTENT_TID.getBytes());
				long tid = bs == null ? 0 :Long.parseLong(new String(bs));
				bs = columnMap.get(HBaseContent.CONTENT_UID.getBytes());
				long uid = bs == null ? 0 :Long.parseLong(new String(bs));
				bs = columnMap.get(HBaseContent.CONTENT_GREAT.getBytes());
				int great = bs == null ? 0 :Integer.parseInt(new String(bs));
				bs = columnMap.get(HBaseContent.CONTENT_PID.getBytes());
				long pid = bs == null ? 0 :Long.parseLong(new String(bs));
				bs = columnMap.get(HBaseContent.TIMESTAMP.getBytes());
				long time = bs == null ? 0 :Long.parseLong(new String(bs));
				TopicContent topicContent = new TopicContent(contentId, contentString, tid, uid, time, great, pid);
				return topicContent;
			} catch (UnsupportedEncodingException e) {
				e.printStackTrace();
				return null;
			}
			}else
				return null;
	}
	/**
	 * 保存用户赞的回复贴，可以是赞，也可以是取消赞
	 * @param contentId 回复帖id
	 * @param type 1,为赞，0为取消赞
	 * @return true/false
	 */
	public void updateTopicContentGreat(long topicId,long contentId,int type) throws IOException{
		long sortKey = Integer.MAX_VALUE - contentId;
		String number = HBaseUtil.numberChange(10, sortKey);
		String rowKey = ""+topicId+number;
		
		String value = hBaseRepository.getValue(HBaseContent.HBASE_TABLE_CONTENT, rowKey, HBaseContent.HBASE_FAMILY_CONTENT	, HBaseContent.CONTENT_GREAT);
		int greatTime = Integer.parseInt(value);
		if(type == 1){
			greatTime ++;
		}else greatTime --;
		hBaseRepository.updateQualifierValue(HBaseContent.HBASE_TABLE_CONTENT, rowKey, HBaseContent.HBASE_FAMILY_CONTENT	, HBaseContent.CONTENT_GREAT, greatTime+"");
	}
	
}